package streamAPI.operators;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import streamAPI.source.MyNoParallelSource;

/**
 * @author yue.cao
 * @since 09-29-2020
 */
public class DemoWithSideOutput {

	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		DataStreamSource<Long> source = env.addSource(new MyNoParallelSource());

		OutputTag<Long> jiShuLiu = new OutputTag<Long>("jiShuLiu"){};
		OutputTag<Long> ouShuLiu = new OutputTag<Long>("ouShuLiu"){};

		SingleOutputStreamOperator<Long> sideOutputData = source.process(new ProcessFunction<Long, Long>() {
			@Override
			public void processElement(Long value, Context ctx, Collector<Long> out) throws Exception {
				if (value % 2 == 0) {
					ctx.output(ouShuLiu, value);
				} else {
					ctx.output(jiShuLiu, value);
				}
			}
		});


		sideOutputData.getSideOutput(jiShuLiu).print().setParallelism(1);
		env.execute();


	}
}
